package cc.fs.androidcode.utils.mqtt;


import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.Listener;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

import java.net.URISyntaxException;


/**
 * Created by fostion on 2015/7/16.
 */
public class MqttClient {
    private static String DKEEPLIVE = "/KeepLive";
    private static final int MAX_RECONNECT = 5;
    private static String TOPICHEAD = "fostion/test";
    private String VERSION = "v1";
    //    private String USERNAME = "yang";
//    private String PASSWORD = "movie_together_mqtt_server";
    private final String HOST = "mosquitto.org";
    private final int PORT = 1883;
    private String topic;
    public static int reconnectTime = 5;
    private final short MQTT_KEEP_ALIVE = 60 * 15;
    private final long MAX_CONNECT_ATTEMPT = 20;
    private final long RECONNECT_DELAY_TIME = 20;
    private final int BUFFER_SIZE = 2048;
    private MQTT mqtt;
    private CallbackConnection callbackConnection;
    public Response response;
//    private String id;

    public MqttClient() {
        try {
//            this.id = _id;
            topic = TOPICHEAD;
            setUp();
        } catch (Exception e) {
            return;
        }
    }

    private void setUp() throws URISyntaxException {
        mqtt = new MQTT();
        //faile reconnect
//        mqtt.setUserName(USERNAME);
//        mqtt.setPassword(PASSWORD);
//        mqtt.setClientId(id);
//        mqtt.setCleanSession(false);
        mqtt.setKeepAlive(MQTT_KEEP_ALIVE);
        mqtt.setConnectAttemptsMax(MAX_CONNECT_ATTEMPT);
        mqtt.setReconnectAttemptsMax(MAX_CONNECT_ATTEMPT);
        mqtt.setReconnectDelay(RECONNECT_DELAY_TIME);
//        //send recive size
        mqtt.setSendBufferSize(BUFFER_SIZE);
        mqtt.setReceiveBufferSize(BUFFER_SIZE);
//
//        //set will
        mqtt.setWillTopic("willTopic");
        mqtt.setWillMessage("willMessage");
        mqtt.setWillQos(QoS.EXACTLY_ONCE);

        mqtt.setHost(HOST, PORT);
        callbackConnection = mqtt.callbackConnection();
        //add listener
        callbackConnection.listener(new Listener() {

            public void onDisconnected() {
            }

            public void onConnected() {
            }

            //get publish
            public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
                try {
                    if (response != null) {
                        String data = new String(payload.toByteArray(), "utf-8");
                        response.run(data);
                    }
                    ack.run();//back server
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }

            public void onFailure(Throwable value) {

            }
        });

        //connect
        callbackConnection.connect(new org.fusesource.mqtt.client.Callback<Void>() {
            public void onFailure(Throwable value) {
                try {
                    //reconnect
                    if (reconnectTime > 0) {
                        setUp();
                        reconnectTime -= reconnectTime;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            // Once we connect..
            public void onSuccess(Void v) {
                //订阅消息
                Topic[] topics = {new Topic(topic, QoS.EXACTLY_ONCE)};// QoS.QoS.EXACTLY_ONCE
                callbackConnection.subscribe(topics, new org.fusesource.mqtt.client.Callback<byte[]>() {
                    public void onSuccess(byte[] qoses) {
                    }

                    public void onFailure(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });
            }
        });
    }

    /**
     * 发布消息
     * @param topic 通道
     * @param msg  消息内容
     * @param callback
     */
    public void publish(String topic, String msg, final Response<Boolean> callback) {
        if(callbackConnection == null) return;
        callbackConnection.publish(topic, msg.getBytes(), QoS.EXACTLY_ONCE, false, new org.fusesource.mqtt.client.Callback<Void>() {
            @Override
            public void onSuccess(Void v) {
                callback.run(true);
            }

            @Override
            public void onFailure(Throwable throwable) {
                callback.run(false);
                throwable.printStackTrace();
            }
        });
    }

    /**
     * 订阅消息
     * @param topic 订阅消息通道
     * @param mCallback 返回
     */
    public void subscribe(String topic, final Response<Boolean> mCallback) {
        if(callbackConnection == null) return;
        //订阅消息
        Topic[] topics = {new Topic(topic, QoS.EXACTLY_ONCE)};
        callbackConnection.subscribe(topics, new org.fusesource.mqtt.client.Callback<byte[]>() {
            public void onSuccess(byte[] qoses) {
                try {
                    mCallback.run(true);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            public void onFailure(Throwable value) {
                mCallback.run(false);
            }
        });
    }

    /**
     * 取消订阅
     */
    public void removeSubscribe() {
        if(callbackConnection == null) return;
        callbackConnection.unsubscribe(new UTF8Buffer[]{UTF8Buffer.utf8(topic)}, new Callback<Void>() {
            @Override
            public void onSuccess(Void aVoid) {
            }

            @Override
            public void onFailure(Throwable throwable) {
            }
        });
    }

    /**
     * 心跳ping
     */
    public void ping(final Response<Boolean> callback) {
        if(callbackConnection == null) return;
        //实现心跳的方式是每隔一段时间就会发一次心跳连接若是失败就重连
        if (callbackConnection != null) {
            callbackConnection.publish(DKEEPLIVE, "alive".getBytes(), QoS.EXACTLY_ONCE, false, new org.fusesource.mqtt.client.Callback<Void>() {
                @Override
                public void onSuccess(Void v) {
                    callback.run(true);
                }

                @Override
                public void onFailure(Throwable throwable) {
                    callback.run(false);
                }
            });
        } else {
            callback.run(false);
        }
    }

    /**
     * 订阅消息
     */
    public void reSubcribe(final Response<Boolean> callback){
        Topic[] topics = {new Topic(topic, QoS.EXACTLY_ONCE)};
        callbackConnection.subscribe(topics, new org.fusesource.mqtt.client.Callback<byte[]>() {
            public void onSuccess(byte[] qoses) {
                callback.run(true);
            }

            public void onFailure(Throwable value) {
                callback.run(false);
            }
        });
    }

    /**
     * 重连
     */
    public void reConnect(final Response<Boolean> callback) {
        if(callbackConnection == null) return;
        reconnectTime = MAX_RECONNECT;
        callbackConnection.connect(new org.fusesource.mqtt.client.Callback<Void>() {
            public void onFailure(Throwable value) {
                try {
                    //reconnnect
                    if (reconnectTime > 0) {
                        setUp();
                        reconnectTime -= reconnectTime;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            // Once we connect..
            public void onSuccess(Void v) {
                //重新订阅消息
                if(callbackConnection == null) return;
                Topic[] topics = {new Topic(topic, QoS.EXACTLY_ONCE)};// QoS.QoS.EXACTLY_ONCE
                callbackConnection.subscribe(topics, new org.fusesource.mqtt.client.Callback<byte[]>() {
                    public void onSuccess(byte[] qoses) {
                    }

                    public void onFailure(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });
            }
        });
    }

    public void setResponse(Response response) {
        this.response = response;
    }

    public void close() {
        if (callbackConnection != null) {
            callbackConnection.disconnect(null);
        }
    }

    public interface Response<T> {
        void run(T param);
    }
}
